//package com.hcy.business.service.mqservice;
//
//import com.hcy.business.service.impl.UserServiceBzImpl;
//import com.hcy.orderclient.OrderBaseClient;
//import com.hcy.ordermodel.dto.request.PayOrderRequestDTO;
//import com.hcy.ordermodel.dto.response.PayOrderResponseDTO;
//import com.hcy.rabbitbasicmodel.enums.Code;
//import com.hcy.rabbitbasicmodel.model.ResponseData;
//import com.hcy.userdomin.domain.service.impl.CustomerInfoServiceImpl;
//import com.mysql.cj.util.StringUtils;
//import org.apache.commons.lang3.builder.ToStringBuilder;
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
//import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.boot.ApplicationArguments;
//import org.springframework.boot.ApplicationRunner;
//import org.springframework.stereotype.Service;
//
//import java.math.BigDecimal;
//
///**
// * @Author huchenying
// * @Description
// * @Date 2022/4/6 22:41
// */
//@Service
//public class MQConsumerChangeMoney implements ApplicationRunner {
//
//    private static final Logger logger = LoggerFactory.getLogger(MQConsumerChangeMoney.class);
//
//    @Value("${MQ.namesrvAddr}")
//    private String addr;
//
//    @Autowired
//    UserServiceBzImpl userServiceBz;
//
//
//    @Override
//    public void run(ApplicationArguments args) throws Exception {
//        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("moneyChange");
//        //设置NameServer地址
//        consumer.setNamesrvAddr(addr);
//        //设置起始消费位置
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//        //指定消费Topic和Tag
//        consumer.subscribe("moneyChange", "*");
//        //指定消费模式 默认集群模式
//        consumer.setMessageModel(MessageModel.CLUSTERING);
//        consumer.setConsumeThreadMin(1);
//        consumer.setConsumeThreadMax(1);
//        consumer.setConsumeMessageBatchMaxSize(8);
//        //注册消息监听器
//        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
//            //逐条消费消息
//            for (MessageExt msg : msgs) {
//                String payNumber = msg.getKeys();
//                consumer(payNumber);
//            }
//            //返回消费状态
//            //重新消费
//            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//        });
//        //开启消费
//        consumer.start();
//        logger.info("websocket mq consumer started");
//    }
//
//    public void consumer(String payNumber) {
//        logger.info("consumer start param -> payNumber = {}", payNumber);
//        try {
//            if (StringUtils.isNullOrEmpty(payNumber)) {
//                throw new RuntimeException("payNumber is null or empty !");
//            }
//            //更改
//            userServiceBz.consumerPay(payNumber);
//        } catch (Exception e) {
//            logger.error("cancel order fail e -> {}", ToStringBuilder.reflectionToString(e));
//        }
//    }
//}
